package main

import (
	"errors"
	"../common"
	"strings"
	"sync"
	"syscall"
	"time"
)

type Sender struct {
	conn               *common.Conn
	dpch               chan common.DataPoint
	stopping           chan struct{}
	stopped            chan struct{}
	metrics            []string
	collector          Collector
	mu                 sync.RWMutex
	send_error         []bool
	CarbonAddrs        []string
	checkAliveInterval int
	retryTimes         int
}

func (s *Sender) addResult(e bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if len(s.send_error) > MAX_SENDER_RESULT_CACHE {
		s.send_error = []bool{}
	}
	s.send_error = append(s.send_error, e)
}

func (s *Sender) HasError() bool {
	s.mu.Lock()
	defer func() {
		s.send_error = []bool{}
		s.mu.Unlock()
	}()
	for _, v := range s.send_error {
		if v == false {
			return false
		}
	}
	return true
}

func NewSender(conf *Config) (Sender, error) {
	var timeOffset int64

	ip, err := common.GetAddr(conf.MondoAddr)
	if err != nil {
		return Sender{}, err
	}
	metricHead := "golia." + strings.Replace(ip, ".", "_", -1)
	dpch := make(chan common.DataPoint, conf.RetryTimes*200)

	//conn, err := NewConn(conf.CarbonAddr)
	conn, err := common.NewConnFrom(conf.CarbonAddrs)
	if err != nil {
		return Sender{}, err
	}
	if conn == nil {
		return Sender{}, errors.New("connection is nil")
	}

	timeOffset, err = common.GetDelta(conf.MondoAddr + "/utils/timestamp")
	if err != nil {
		log.Info("can't get time offset ,use 0")
	}
	collector := Collector{
		ch:              dpch,
		metricHead:      metricHead,
		interval:        conf.MetricInterval,
		timeOffset:      timeOffset,
		diskMetricTimes: conf.DiskMetricTimes,
	}
	sender := Sender{
		metrics:            conf.Metrics,
		conn:               conn,
		dpch:               dpch,
		collector:          collector,
		CarbonAddrs:        conf.CarbonAddrs,
		checkAliveInterval: conf.CheckAliveInterval,
		retryTimes:         conf.RetryTimes,
	}
	return sender, nil
}

func (s *Sender) Run() {
	s.stopping = make(chan struct{})
	s.stopped = make(chan struct{})
	ticker := time.NewTicker(time.Duration(s.checkAliveInterval) * time.Second)
	//var err error
	go s.collector.CollectAllMetric(s.metrics)
	defer s.collector.Stop()
	for {
		select {
		case <-ticker.C:
			if s.retryTimes < 1 {
				KillSelf(syscall.Getpid(), syscall.SIGTERM)
				//syscall.Kill(syscall.Getpid(), syscall.SIGTERM)
			}
			if !s.conn.IsAlive() {
				//if newConn, err := NewConn(s.CarbonAddr); err != nil {
				if newConn, err := common.NewConnFrom(s.CarbonAddrs); err != nil {
					s.retryTimes--
					log.Warningf("failed to reconnect, retry times remain: %d, err is:%v", s.retryTimes, err)
					time.Sleep(time.Duration(5) * time.Second)
				} else {
					s.conn = newConn
					s.retryTimes++
				}
			}
		case dp := <-s.dpch:
			if n, err := s.conn.WriteDataPoint(dp); err != nil || n == 0 {
				log.Warningf("can't send metric %v", dp)
				s.addResult(false)
			} else {
				log.Debugf("successfully send metric %v", dp)
				s.addResult(true)
			}
		case <-s.stopping:
			log.Info("shut down sender")
			s.conn.Close()
			close(s.stopped)
			return
		}
	}
}

func (s *Sender) Stop() {
	close(s.stopping)
	<-s.stopped
}
